package com.danbay.framework.kafka;

import com.danbay.framework.core.KeyValuePair;
import com.danbay.framework.kafka.coap.CoAPPackageBuilder;
import com.danbay.framework.redis.RedisClient;
import com.danbay.framework.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.californium.core.coap.CoAP;
import org.eclipse.californium.core.coap.Request;
import org.eclipse.californium.core.coap.Response;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;

/**
 * @author SJG
 *         2017/11/14.
 */
@Slf4j
@EnableConfigurationProperties
public class AccessKafkaTopic {

    /**
     * 存放kafka消息coap协议中的mid与命令uri path对应的键
     */
    private static final String KAFKA_COAP_MID_KEY = "kafka_CoAP_mid:";
    /**
     * 存放kafka消息id的键
     */
    private static final String KAFKA_MESSAGE_ID_KEY = "kafka_message_id:";

    @Value("${coap.mid.expires:300}")
    private int coap_mid_expires = 300;

    /**
     * 消息类型
     */
    private int messageType;

    /**
     * 主题名
     */
    private String targetTopic;

    /**
     * 消息来源
     */
    private String sourceTopic;

    private KafkaClient kafkaClient;

    private RedisClient redisClient;

    public AccessKafkaTopic(String targetTopic, String sourceTopic, byte messageType, KafkaClient kafkaClient, RedisClient redisClient) {
        this.targetTopic = targetTopic;
        this.sourceTopic = sourceTopic;
        this.messageType = messageType;
        this.kafkaClient = kafkaClient;
        this.redisClient = redisClient;
    }

    public void send(Object data){
        AccessPackage<Object> accessPackage = new AccessPackage<>();
        accessPackage.setHeaders(messageType, MessageSequence.getNextSequence(), System.currentTimeMillis(), targetTopic);
        accessPackage.setData(data);
        kafkaClient.send(targetTopic, accessPackage);
    }

    /**
     * 发布消息
     *
     * @param request 数据包
     */
    private void send(Request request) {
        send(request, MessageSequence.getNextSequence());
    }

    /**
     * 发布消息
     *
     * @param request 数据包
     */
    private void send(Request request, Long srcSequence) {
        if (request.getType() == CoAP.Type.CON) {

        }

        AccessPackage<Request> accessPackage = new AccessPackage<>();
        accessPackage.setHeaders(messageType, srcSequence, System.currentTimeMillis(), targetTopic);
        accessPackage.setData(request);
        kafkaClient.send(targetTopic, accessPackage);
    }

    /**
     * 发送响应包
     *
     * @param cmd  目标命令url
     * @param data 数据
     * @param <T>  数据类型
     */
    public <T> void sendResponse(String cmd, T data) {
        Response response = CoAPPackageBuilder.buildPackage(cmd, data);
        send(response);
    }

    /**
     * 发送响应包
     *
     * @param response 数据包
     */
    public void send(Response response) {
        send(response, MessageSequence.getNextSequence());
    }

    /**
     * 发送响应包
     *
     * @param response 数据包
     */
    public void send(Response response, Long sequence) {
        AccessPackage<Response> accessPackage = new AccessPackage<>();
        accessPackage.setHeaders(messageType, sequence, System.currentTimeMillis(), sourceTopic);
        accessPackage.setData(response);
        kafkaClient.send(targetTopic, accessPackage);
    }

    /**
     * 发送kafka ACK 响应包
     *
     * @param srcSequence 源sequence
     */
    public void sendACK(Long srcSequence) {
        AccessPackage<String> accessPackage = new AccessPackage<>();
        accessPackage.setHeaders(messageType, srcSequence, System.currentTimeMillis(), sourceTopic);
        kafkaClient.send(targetTopic, accessPackage);
    }

    /**
     * post方式发布不需要响应的消息
     *
     * @param cmd  目标命令url
     * @param data 数据
     * @param <T>  数据类型
     */
    public <T> void postNON(String cmd, T data) {
        send(CoAPPackageBuilder.buildPostNONPackage(cmd, data));
    }

    /**
     * post方式发布不需要响应的消息
     *
     * @param cmd         目标命令url
     * @param srcSequence 源消息sequence
     * @param data        数据
     * @param <T>         数据类型
     */
    public <T> void postNON(String cmd, Long srcSequence, T data) {
        send(CoAPPackageBuilder.buildPostNONPackage(cmd, data), srcSequence);
    }

    /**
     * post方式发布不需要响应的消息，不带数据
     *
     * @param cmd 目标命令url
     */
    public void postNON(String cmd) {
        postNON(cmd, null);
    }


    /**
     * ack响应请求的消息
     *
     * @param srcSequence 源sequence
     */
    public void ack(Long srcSequence) {
        sendACK(srcSequence);
    }

    /**
     * ack回应请求的消息
     *
     * @param cmd      命令url
     * @param srcMid   源mid
     * @param srcToken 源token
     */
    public <T> void ackData(String cmd, int srcMid, Long srcToken, T data) {
        send(CoAPPackageBuilder.buildACKDataPackage(cmd, srcMid, srcToken, data));
    }

    /**
     * ack回应错误的请求消息
     *
     * @param cmd      命令url
     * @param srcMid   源mid
     * @param srcToken 源token
     */
    public void ackBadRequest(String cmd, int srcMid, Long srcToken) {
        send(CoAPPackageBuilder.buildACKBadRequestPackage(cmd, srcMid, srcToken));
    }

    public void ackBadRequest(CoAP.ResponseCode responseCode,String cmd, int srcMid, Long srcToken){
        send(CoAPPackageBuilder.buildPackage(responseCode,cmd, srcMid, srcToken, null));
    }

    /**
     * post方式发布需要响应的消息，可分别指定payload和uriQuery数据
     *
     * @param cmd      目标命令url
     * @param data     payload数据
     * @param uriQuery uriQuery数据
     * @param <T>      数据类型
     * @return 消息id
     */
    public <T> String postCON(String cmd, T data, KeyValuePair<String>... uriQuery) {
        String uriPath = cmd;
        if (uriQuery != null && uriQuery.length > 0) {
            //根据uriQuery生成cmd
            cmd = generateCmd(cmd, uriQuery);
        }
        Request request = CoAPPackageBuilder.buildPostCONPackage(cmd, data);
        int mid = request.getMID();
        //先缓存mid,以便客户端ack时找到对应的命令
        redisClient.set(getACoAPMidUriPathKey(mid), uriPath, coap_mid_expires);
        //发送
        send(request);
        //生成并缓存请求消息ID，收到回应消息时将原请求消息ID携带回业务方法，以便业务方法可以找到原请求消息的数据
        String messageId = StringUtils.newUUID();
        redisClient.set(getKafkaMessageIdKey(mid), messageId);
        log.info("store messageId:{},mid:{}", messageId, mid);
        return messageId;
    }

    /**
     * 使用uriQuery数据，以post方式发布需要响应的消息
     *
     * @param cmd 目标命令url
     * @param kvs uri query中的数据
     * @return 消息id
     */
    public String postCONWithUriQuery(String cmd, KeyValuePair<String>... kvs) {
        return postCON(cmd, null, kvs);
    }

    /**
     * 根据uriQuery生成cmd
     *
     * @param cmd 原始cmd
     * @param kvs uriQuery
     * @return 新的cmd
     */
    private String generateCmd(String cmd, KeyValuePair<String>... kvs) {
        //组合uri query查询参数串
        if (kvs != null && kvs.length > 0) {
            String uriQuery = "";
            for (KeyValuePair keyValuePair : kvs) {
                uriQuery += keyValuePair.getKey() + "=" + keyValuePair.getValue() + "&";
            }
            //del last &
            uriQuery = uriQuery.substring(0, uriQuery.length());
            cmd = cmd + "?" + uriQuery;
        }
        return cmd;
    }

    /**
     * post方式发布需要响应的消息，不带数据
     *
     * @param cmd 目标命令url
     */
    public void postCON(String cmd) {
        postCON(cmd, null);
    }

    /**
     * get方式发布需要响应的消息
     *
     * @param cmd  目标命令url
     * @param data 数据
     * @param <T>  数据类型
     */
    public <T> void getCON(String cmd, T data) {
        send(CoAPPackageBuilder.buildGetCONPackage(cmd, data));
    }

    /**
     * 发送get请求需要回应的数据包
     *
     * @param cmd      目标命令url
     * @param data     数据
     * @param uriQuery uri query中的数据
     * @param <T>      数据类型
     */
    public <T> void getCON(String cmd, T data, KeyValuePair<String>... uriQuery) {
        String uriPath = cmd;
        if (uriQuery != null && uriQuery.length > 0) {
            //根据uriQuery生成cmd
            cmd = generateCmd(cmd, uriQuery);
        }
        Request request = CoAPPackageBuilder.buildGetCONPackage(cmd, data);
        int mid = request.getMID();
        //先缓存mid,以便客户端ack时找到对应的命令
        redisClient.set(getACoAPMidUriPathKey(mid), uriPath, coap_mid_expires);
        send(request);
    }

    /**
     * get方式发布需要响应的消息，不带数据
     *
     * @param cmd 目标命令url
     */
    public void getCON(String cmd) {
        getCON(cmd, null);
    }


    /**
     * 获取kafka消息coap协议中的mid与命令uri path存放在redis中的键
     *
     * @param mid mid
     * @return redis key
     */
    public static String getACoAPMidUriPathKey(int mid) {
        return KAFKA_COAP_MID_KEY + mid;
    }

    /**
     * 存放kafka消息id的键
     *
     * @param mid mid
     * @return redis key
     */
    public static String getKafkaMessageIdKey(int mid) {
        return KAFKA_MESSAGE_ID_KEY + mid;
    }

    /**
     * 获取coap协议中mid对应的原coap请求中的uri path
     *
     * @param mid mid
     * @return uri path
     */
    public String getCoAPUriPathByMid(int mid) {
        return redisClient.get(getACoAPMidUriPathKey(mid));
    }

    /**
     * 通过mid获取kafka消息
     *
     * @param mid mid
     * @return messageId
     */
    public String getKafkaMessageIdByMid(int mid) {
        return redisClient.get(getKafkaMessageIdKey(mid));
    }
}
